package com.atguigu.education.app.dwd.db;

import com.atguigu.education.util.KafkaUtil;
import com.atguigu.education.util.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTestExam {
    public static void main(String[] args){
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */
        // TODO 3 从Kafka读取业务数据，封装成Flink SQL 表
        String group_id="dwd_test_exam";
        tableEnv.executeSql("create table topic_db(" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`data` map<string, string>,\n" +
                "`pt` AS PROCTIME(),\n" +
                "`ts` string\n" +
                ")" + KafkaUtil.getKafkaDDL("topic_db", group_id));


        // TODO 4 读取测试表数据，建立look-up join表
        Table testExam = tableEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['paper_id'] paper_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['score'] score,\n" +
                "data['duration_sec'] duration_sec,\n" +
                "data['create_time'] create_time,\n" +
                "data['submit_time'] submit_time,\n" +
                "pt,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where `table` = 'test_exam'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("test_exam", testExam);

        Table testPaper = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['course_id'] course_id,\n" +
                " pt,\n" +
                " ts\n" +
                "from topic_db\n" +
                "where `table` = 'test_paper'\n");
        tableEnv.createTemporaryView("test_paper", testPaper);
        // TODO 6 关联两张表
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "te.id,\n" +
                "te.paper_id,\n" +
                "te.user_id,\n" +
                "te.score,\n" +
                "te.duration_sec,\n" +
                "te.create_time,\n" +
                "te.submit_time,\n" +
                "tp.course_id,\n" +
                "te.ts\n" +
                "from test_exam as te\n" +
                "left join\n" +
                "test_paper as tp\n" +
                "on te.paper_id = tp.id");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 7 建立Kafka-Connector dwd_test_exam 表
      tableEnv.executeSql("create table dwd_test_exam(\n" +
                "id string,\n" +
                "paper_id string,\n" +
                "user_id string,\n" +
                "score string,\n" +
                "duration_sec string,\n" +
                "create_time string,\n" +
                "submit_time string,\n" +
                "course_id string,\n" +
                "ts string\n" +
                ")" + KafkaUtil.getKafkaSinkDDL("dwd_test_exam"));
        // TODO 8 将关联结果写入Kafka-Connector 表
        tableEnv.executeSql("insert into dwd_test_exam select * from result_table");
    }
}
